package cn.edu.thu.tsquality.core.common.datasource;

import cn.edu.thu.tsquality.core.common.table.*;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.csv.CSVRecord;

import java.io.*;
import java.security.InvalidParameterException;
import java.util.*;
import java.util.regex.Pattern;

/**
 * Created by wangyihan  on 2018/4/2 下午5:31. E-mail address is yihanwang22@163.com. Copyright © 2017
 * wangyihan. All Rights Reserved.
 *
 * @author wangyihan
 */
public class CsvDataSource implements IDataSource {

  private final String filename;
  private final Reader reader;
  private final CSVParser parser;
  private Header header;
  private int curid;
  private Iterator<CSVRecord> iterator;

  public CsvDataSource(String filename) throws IOException {
    this.filename = filename;
    if (filename == null) {
      throw new InvalidParameterException("Empty filename for CSV source");
    }
    File file = new File(filename);
    if (!file.exists()) {
      this.reader = null;
      this.parser = null;
    } else {
      this.reader = new FileReader(filename);
      this.parser = new CSVParser(this.reader, CSVFormat.RFC4180.withFirstRecordAsHeader());
      this.header = new Header(this.parser.getHeaderMap());
      this.iterator = parser.iterator();
      this.curid = 0;
    }
  }

  public StreamingTable createTable() throws IOException {
    return new StreamingTable(this, this.header);
  }

  public Row readRow() {
    if (!this.iterator.hasNext()) {
      try {
        this.reader.close();
        this.parser.close();
      } catch (IOException ignored) {
      }
      return null;
    }
    CSVRecord csvRecord = this.iterator.next();
    List<Object> rowData = new ArrayList<>();
    for (String attr : this.header.toArray()) {
      rowData.add(csvRecord.get(attr));
    }
    Row rlt = new Row(this.header, rowData, curid);
    curid++;
    return rlt;
  }

  public boolean hasNext() {
    return this.iterator.hasNext();
  }

  @Override
  public void filter(String[] attrs) {
    Set<String> set = new HashSet<>();
    Collections.addAll(set, attrs);
    for (String attr : this.header.toArray()) {
      if (!set.contains(attr)) {
        this.header.remove(attr);
      }
    }
  }

  public Table read() throws IOException {
    if (this.reader == null || this.parser == null) {
      throw new IllegalStateException(
          "Cannot read DataSource, since this file does not exist:" + this.filename);
    }
    int tmpid = 0;
    List<Row> rowList = new ArrayList<>();
    for (CSVRecord record : this.parser.getRecords()) {
      List<Object> rowData = new ArrayList<>();
      for (String attr : this.header.toArray()) {
        if (this.header.isWithDataType()) {
          DataType dt = this.header.getDataType(this.header.getIndex(attr));
          rowData.add(toObject(record.get(attr), dt));
        } else {
          rowData.add(record.get(attr));
        }
      }
      rowList.add(new Row(this.header, rowData, tmpid));
      tmpid++;
    }
    this.reader.close();
    this.parser.close();
    return new Table(this.header, rowList);
  }

  public void write(Table table) throws IOException {
    String[] attrs = table.getHeader().toArray();
    List<Row> rowList = table.getRows();
    File dest = new File(this.filename);
    if (!dest.getParentFile().exists()) {
      if (!dest.getParentFile().mkdirs()) {
        throw new IllegalStateException(
            "Cannot create directory: " + dest.getParentFile().toString());
      }
    }
    Writer out = new FileWriter(dest);
    CSVPrinter printer = new CSVPrinter(out, CSVFormat.RFC4180.withHeader(attrs));
    for (Row row : rowList) {
      printer.printRecord(row.toArray());
    }
    out.close();
    printer.close();
  }

  public Object string2Object(String str) {
    final String intRegx = "^[+-]?\\d+";
    final Pattern intPattern = Pattern.compile(intRegx);
    final String doubleRegx = "^([+-]?\\d+)(\\.\\d+)";
    final Pattern doublePattern = Pattern.compile(doubleRegx);
    if (intPattern.matcher(str).matches()) {
      try {
        Integer i = Integer.valueOf(str);
        return ((Object) i);
      } catch (Exception ex) {
        return ((Object) str);
      }
    } else if (doublePattern.matcher(str).matches()) {
      try {
        Double d = Double.valueOf(str);
        return ((Object) d);
      } catch (Exception ex) {
        return ((Object) str);
      }
    } else {
      return ((Object) str);
    }

  }

  public Object toObject(String str, DataType dt) {
    Object rlt = null;
    switch (dt) {
      case STRING:
        rlt = str;
        break;
      case INTEGER:
        rlt = new Integer(Integer.parseInt(str));
        break;
      case FLOAT:
        rlt = new Float(Float.parseFloat(str));
        break;
      case DOUBLE:
        rlt = new Double(Double.parseDouble(str));
        break;
      default:
        throw new IllegalArgumentException("String cannot be parsed: " + str);
    }
    return rlt;
  }
}
